package cn.itcast.flink.base;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Author itcast
 * Date 2021/7/27 15:13
 * 将两个String类型的流进行union
 * 将一个String类型和一个Long类型的流进行connect
 */
public class UnionConnectTransformation {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setParallelism(1);

        //2.Source
        DataStream<String> ds1 = env.fromElements("zhangsan", "male", "24");
        DataStream<String> ds2 = env.fromElements("wangwu", "female", "25");
        DataStream<Long> ds3 = env.fromElements(80L, 85L, 100L);

        //3. 实现一个 union 操作
        DataStream<String> unionDataStream = ds1.union(ds2);
        //3.2 实现一个connect 操作
        ds1.connect(ds3)
                .process(new CoProcessFunction<String, Long, String>() {
                    @Override
                    public void processElement1(String value, Context ctx, Collector<String> out) throws Exception {
                        out.collect("string value: "+value);
                    }

                    @Override
                    public void processElement2(Long value, Context ctx, Collector<String> out) throws Exception {
                        out.collect("long value:"+value);
                    }
                }).print("connect value: ");

        //4.打印输出
        unionDataStream.print();

        //5.执行环境
        env.execute();
    }
}
